package com.peng.my_rocketmq.consumer;

import java.io.IOException;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.peng.my_rocketmq.bean.User;

@Service
@RocketMQMessageListener
(topic = "peng-mq-spring-test", 
selectorExpression = "syncTagSend3",
consumerGroup = "${spring.application.name}-syncTagSend")
public class SyncMessageExtConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {   


    // 实现消息的消费处理

    @Autowired
	private ObjectMapper objectMapper;
    public void onMessage(MessageExt  messageExt) {
    	String key=messageExt.getKeys();
		String body=new String(messageExt.getBody());
		try {
			User message=objectMapper.readValue(messageExt.getBody(), User.class);
			//User message=(User) JSONObject.parse(body);
	        System.out.printf("接受到------- :%s, body:%s key %s %n", message.getId(), message.getUsername(),key);   
    
		} catch (IOException e) {
			// TODO Auto-generated catch block
			//e.printStackTrace();
		}
		
    	
    }    

   

    // 设置从当前时间点开始消费消息  

   @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);       
       consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));   

   }

}

